package drds.binlog.client.kafka;

import com.google.protobuf.ByteString;
import drds.binlog.protocol.Message;
import drds.binlog.protocol.entry_protocol_proto.Entry;
import drds.binlog.protocol.exception.ClientException;
import drds.binlog.protocol.protocol_proto.Compression;
import drds.binlog.protocol.protocol_proto.Messages;
import drds.binlog.protocol.protocol_proto.Packet;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

/**
 * Kafka Message类的反序列化
 */
public class MessageDeserializer implements Deserializer<Message>
{

    private static Logger logger = LoggerFactory.getLogger(MessageDeserializer.class);

    @Override
    public void configure(Map<String, ?> configs, boolean isKey)
    {
    }

    @Override
    public Message deserialize(String topic, byte[] bytes)
    {
        try
        {
            if (bytes == null)
            {
                return null;
            } else
            {
                Packet packet = Packet.parseFrom(bytes);
                switch (packet.getType())
                {
                    case messages:
                    {
                        if (!packet.getCompression().equals(Compression.none))
                        {
                            throw new ClientException("compression is not supported in this connector");
                        }

                        Messages messages = Messages.parseFrom(packet.getBody());
                        Message message = new Message(messages.getBatchId());
                        for (ByteString byteString : messages.getMessagesList())
                        {
                            message.addEntry(Entry.parseFrom(byteString));
                        }
                        return message;
                    }
                    default:
                        break;
                }
            }
        } catch (Exception e)
        {
            logger.error("Error when deserializing byte[] to message ");
        }
        return null;
    }

    @Override
    public void close()
    {
        // nothing to do
    }
}
